package server

import (
	"context"
	"core/corestruct"
	"core/datetime"
	mymongo "core/mongo"
	"core/server/baseserver"
	"core/server/kafkaserver"
	"encoding/json"
	"fmt"
	"iotall/iotcore/entity"
	"time"

	"github.com/dubbogo/gost/log/logger"
	kafkago "github.com/segmentio/kafka-go"
	"go.mongodb.org/mongo-driver/bson"
	"go.mongodb.org/mongo-driver/mongo"
)

type eventServer struct {
	kafkaserver.ConsumerServer
	baseserver.BaseServer
	list *corestruct.Queue

	lstDt time.Time
	times int
}

func (p *eventServer) Start() {
	p.ConsumerServer.Name = "警情处理"
	p.ConsumerServer.Topic = entity.EVENTTOPIC
	p.ConsumerServer.GroupID = "event"
	p.ConsumerServer.Do = p.eventServerDo

	p.BaseServer.Name = p.ConsumerServer.Name + "定时处理"
	p.BaseServer.DoInterval = p.DoInterval
	p.BaseServer.DoClose = p.DoInterval
	p.BaseServer.Interval = 5000

	p.lstDt = time.Now()

	p.list = &corestruct.Queue{}
	p.ConsumerServer.Start()
	p.BaseServer.Start()

}
func (p *eventServer) Close() {

}
func (p *eventServer) FindById(id string) *entity.StdPkgDeviceMongo {
	filter := bson.M{"_id": id}
	collection := mymongo.Db.Collection("StdPkgDevice")
	var sr = collection.FindOne(context.TODO(), filter)
	if sr.Err() != nil {
		return nil
	}
	var r = &entity.StdPkgDeviceMongo{}
	sr.Decode(r)
	return r
}

func (p *eventServer) DoInterval() {
	var list = p.list.PopAll()
	if len(list) == 0 {
		return
	}
	var operations []mongo.WriteModel
	for _, record := range list {
		mdev, ok := record.(*entity.StdPkgDeviceMongo)
		if !ok {
			fmt.Println("数据格式不匹配：", record)
			continue
		}

		filter := bson.M{"_id": mdev.Id}
		update := bson.M{"$set": mdev}
		operation := mongo.NewUpdateOneModel()
		operation.SetFilter(filter)
		operation.SetUpdate(update)
		operation.SetUpsert(true) // 如果存在数据执行更新，否则插入
		operations = append(operations, operation)
	}
	collection := mymongo.Db.Collection("StdPkgDevice")
	r, err := collection.BulkWrite(context.Background(), operations)
	if err != nil {
		logger.Error(err)
	} else {
		logger.Infof("InsertedCount:%v  ModifiedCount:%v UpsertedCount:%v \r\n", r.InsertedCount, r.ModifiedCount, r.UpsertedCount)
	}
}
func (p *eventServer) eventServerDo(msg kafkago.Message) error {
	var mpk = &entity.StdPkg{}
	err := json.Unmarshal(msg.Value, mpk)
	if err != nil {
		logger.Info("读取数据错误 :%v", err)
		return err
	}
	for _, v := range mpk.Devices {
		p.times++
		var mdev = &entity.StdPkgDeviceMongo{
			Id:      fmt.Sprintf("%v*%v", mpk.Sid, v.Id),
			Sid:     mpk.Sid,
			Did:     v.Id,
			Name:    v.Name,
			Ability: v.Ability,

			CrtDt: datetime.NowStr(),
		}
		var mdevold = p.FindById(mdev.Id)
		if mdevold == nil {
			fmt.Println("nil")
		}
		p.list.Push(mdev)
	}
	if p.lstDt.Second() != time.Now().Second() {
		logger.Infof("QPS:%v", p.times)
		p.lstDt = time.Now()
		p.times = 0
	}
	//mylogger.Logger.Info("数据",
	//	zap.String("Dt", datetime.FormatDateTimeMs(time.Now())),
	//	zap.String("Data", mpk.GetMemo()),
	//	zap.Int64("Id", mpk.Id),
	//	zap.Int64("Offset", msg.Offset),
	//)
	//2351622
	return nil
}

var (
	esvr *eventServer
)

func init() {
	esvr = &eventServer{}
	//esvr.Start()

}
